package cn.doitedu.demo.stream_table;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;

public class 官方Changelog流转表 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);



        int id = 1;
        String name = "zs";
        int age = 18;
        Row r1 = Row.ofKind(RowKind.UPDATE_BEFORE, id, name, age);

        id = 1;
        name = "zs";
        age = 18;
        Row r2 = Row.ofKind(RowKind.UPDATE_AFTER, id, name, age);

        DataStreamSource<Row> stream = env.fromElements(r1,r2);


        SingleOutputStreamOperator<Row> process = stream.process(new ProcessFunction<Row, Row>() {
            @Override
            public void processElement(Row row, ProcessFunction<Row, Row>.Context context, Collector<Row> collector) throws Exception {
                collector.collect(row);
            }
        });


        Table table = tenv.fromChangelogStream(process);
        tenv.createTemporaryView("xx",table);
        tenv.executeSql("select * from xx").print();

        env.execute();
    }

}
